/*
 * Copyright 2009-2010 by The Regents of the University of California
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * you may obtain a copy of the License from
 * 
 *     http://www.apache.org/licenses/LICENSE-2.0
 * 
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package edu.ucr.cs.dblab.acestore.operators;

import java.nio.ByteBuffer;

import edu.ucr.cs.dblab.acestore.api.PageAccessHelper;
import edu.ucr.cs.dblab.acestore.base.datatypes.SchemaDescriptor;

/**
 * @author jarodwen
 * @created Jun 21, 2011
 */
public class BufferedPageSorter {
    
    private static final int INTSIZE = Integer.SIZE / 8;

    public static void sort(ByteBuffer page, SchemaDescriptor schemaDescriptor, int[] compareFields){
        
    }
    
    /***********************************
     * Quick sort functions
     ***********************************/

    /**
     * Quick function.
     * 
     * Partition the whole data set into two parts using the pivot
     * value, and do quick sort on the two parts separately.
     * 
     * @param links
     * @param low
     * @param high
     * @return
     * @throws Exception
     */
    private static void quicksort(ByteBuffer page, int low, int high, SchemaDescriptor schemaDescriptor, int[] compareFields)
            throws Exception {
        if (high > low) {
            int partitionPivotIndex = (int) (Math.random() * (high - low))
                    / 2 * 2 + low;
            int newPivotIndex = partition(page, low, high,
                    partitionPivotIndex, schemaDescriptor, compareFields);
            quicksort(page, low, newPivotIndex - 2, schemaDescriptor, compareFields);
            quicksort(page, newPivotIndex + 2, high, schemaDescriptor, compareFields);
        }
    }

    private static int partition(ByteBuffer page, int lo, int hi, int pivotIndex, SchemaDescriptor schemaDescriptor, int[] compareFields)
            throws Exception {

        swap(page, pivotIndex, hi); // send to the back

        int index = lo;

        for (int i = lo; i < hi; i = i + 2) {
            if (compareLinks(page, i, hi, schemaDescriptor, compareFields) <= 0) {
                swap(page, i, index);
                index = index + 2;
            }
        }

        swap(page, hi, index);

        return index;
    }

    private static void swap(ByteBuffer page, int i, int j) throws Exception {
        int offset1 = PageAccessHelper.GetRecStartOffset(page, i);
        int len1 = PageAccessHelper.GetRecEndOffset(page, i) + 1 - offset1;
        byte[] tmp1 = new byte[len1];
        page.position(offset1);
        page.get(tmp1);
        
        int offset2 = PageAccessHelper.GetRecStartOffset(page, j);
        int len2 = PageAccessHelper.GetRecEndOffset(page, j) + 1 - offset2;
        if(len2 != len1){
            // FIXME
            throw new Exception("Currently in-mem sorter do not support variable-length records.");
        }
        
        byte[] tmp2 = new byte[len2];
        page.position(offset2);
        page.get(tmp2);
        
        // copy bytes
        page.position(offset1);
        page.put(tmp2);
        
        page.position(offset2);
        page.put(tmp1);
    }

    private static int compareLinks(ByteBuffer page, int i, int j, SchemaDescriptor schemaDescriptor, int[] compareFields)
            throws Exception {
        int offset1 = PageAccessHelper.GetRecStartOffset(page, i);
        int offset2 = PageAccessHelper.GetRecStartOffset(page, j);
        int fieldCount1 = page.getInt(offset1);
        int fieldCount2 = page.getInt(offset2);
        for (int k = 0; k < compareFields.length; k++) {
            int len = schemaDescriptor.getFields()[compareFields[i]]
                    .getTypeSize();

            int cmp = schemaDescriptor.getFields()[compareFields[i]].compare(
                    page.array(), offset1 + INTSIZE + fieldCount1
                            * INTSIZE + compareFields[i] * INTSIZE, len,
                    page.array(), offset2 + INTSIZE + fieldCount2
                            * INTSIZE + compareFields[i] * INTSIZE, len);
            
            if(cmp == 0)
                continue;
            else if(cmp > 0)
                return 1;
            else
                return -1;
        }
        return 0;
    }

    
    /**
     * @param args
     */
    public static void main(String[] args) {
        // TODO Auto-generated method stub

    }

}
